package com.example.demo.maintainer;

import com.example.demo.Application;
import com.example.demo.service.CacheService;
import com.example.demo.service.ChannelService;
import com.example.demo.util.DateUtil;
import com.example.demo.work.GatewayOperator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.sql.Time;
import java.sql.Timestamp;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiFunction;
import java.util.stream.Collectors;

/**
 * 通道维护者
 * 两个职能
 * 1. 接收来自通道竞争者竞争到的通道id, 做初始化或更新通道信息的操作, 并且向通道工作器提交发送任务申请
 * 2. 维护本地通道信息
 * <p>
 * Created by guoyangqiao on 10/05/2017.
 */
@Component
public class GatewayOfficer {
    private static final Logger LOGGER = LoggerFactory.getLogger(GatewayOfficer.class);
    private static final int PARALLELISM_NUM = Runtime.getRuntime().availableProcessors();

    private static GatewayOfficer gatewayOfficer;

    @Resource
    private ChannelService channelService;
    @Resource
    private CacheService cacheService;

    @PostConstruct
    private void init(){
        gatewayOfficer = this;
        gatewayOfficer.channelService = this.channelService;
        gatewayOfficer.cacheService = this.cacheService;
    }

    //通道空闲多久释放发送权限
    private static final long RELEASE_GATEWAY_EXPIRE_MINUTE = 1;
    //最多允许多少个工作线程?
    private static final ExecutorService GLOBAL_GATEWAY_TASK_EXECUTOR = new ThreadPoolExecutor(36, 36, 5, TimeUnit.MINUTES, new LinkedBlockingQueue<>(), new ThreadFactory() {
        private AtomicInteger ac = new AtomicInteger();

        @Override
        public Thread newThread(Runnable r) {
            return new Thread(r, "GATEWAY_EXECUTOR_" + ac.getAndIncrement());
        }
    }, new ThreadPoolExecutor.AbortPolicy());
    //core
    public static final ConcurrentHashMap<Integer, GatewayOperator> AVAILABLE_GATEWAY = new ConcurrentHashMap<>();


    static {
        LOGGER.info("开始启动[定时维护通道的任务]");
        try {
            Class.forName(GatewayMaintainer.class.getName());
        } catch (ClassNotFoundException e) {
            LOGGER.info("不存在的!", e);
        }
        LOGGER.info("[定时维护通道的任务]启动结束");
    }

    /**
     * 竞争通道
     */
    public static boolean ifGatewayPrivilegeExistsThenRefreshPrivilegeTime(int gatewayId) {
        return GatewayOfficer.AVAILABLE_GATEWAY.computeIfPresent(gatewayId, (gid, unit) ->
        {
            unit.setLastUpdateTime();
            return unit;
        }) != null;
    }

    /**
     * 接收发送请求
     */
    public static void acceptNewGatewayWork(final int gatewayId, final String serverName) {
        GatewayOperator operator = AVAILABLE_GATEWAY.computeIfAbsent(gatewayId, (gid) -> {
//            LOGGER.info("{}应用当前没有通道{},新增通道", Application.SERVER_NAME_ID, gatewayId);
            Map<String, Object> gatewayConfig = gatewayOfficer.channelService.getSendConfigIfGatewayValid(gatewayId);
            GatewayOperator gatewayRunTImeUnit;
            if (gatewayConfig != null) {
//                LOGGER.info("通道{}配置[有效],新增通道,配置信息:" + gatewayConfig, gatewayId);
                gatewayRunTImeUnit = GatewayOperator.newInstance(gid, serverName, System.currentTimeMillis(), true, sendTimeAllowed(gatewayConfig), (String) gatewayConfig.get("sendConfig"), (Integer) gatewayConfig.get("limitnum"), (Integer) gatewayConfig.get("threadnum"));
            } else {
//                LOGGER.info("通道{}配置[无效],新增通道,将发送权限设为无效", gatewayId);
                gatewayRunTImeUnit = GatewayOperator.newInstance(gid, serverName, System.currentTimeMillis(), false, false, null, 0, 0);
            }
            return gatewayRunTImeUnit;
        });
        operator.addAcceptedWorkCounter();
        syncSubmitWorkThreadIfPermitted(operator);
    }

    /**
     * 通道任务执行完后的收尾
     */
    public static void afterThreadWorkFinished(GatewayOperator operator, long thatTimeWorkCount) {
        operator.setLastUpdateTime();
        long currentRunningTreads = operator.decreaseCurrentRunningTreads();
//        LOGGER.info("通道{}的线程执行结束,仍在运行的线程数为{}", operator.getGatewayId(), currentRunningTreads);
        if (currentRunningTreads == 0) {//当前线程为最后一个线程
            long acceptedWorkCounter = operator.getAcceptedWorkCounter();
            if (thatTimeWorkCount < acceptedWorkCounter) {//意味着在上次任务触发后,又有更多的任务进来,虽然可能被之前线程处理过了,但是为了保证新的任务能马上被执行到,就算是空转也要看下有没有数据
//                LOGGER.info("当前线程是通道{}的最后线程,线程开始时任务数{},现在任务数{},提交发送任务", operator.getGatewayId(), thatTimeWorkCount, acceptedWorkCounter);
                syncSubmitWorkThreadIfPermitted(operator);
            }
        }
        LOGGER.info("通道{}发送任务结束", operator.getGatewayId());
    }

    //<====================================上面是通道发送相关,下面是通道维护相关====================================>

    /**
     * 专门用来维护通道的类
     */
    private static class GatewayMaintainer {

        static {
            new Thread(() -> {
                while (true) {
                    try {
//                        LOGGER.info("{}开始定时维护应用通道信息. " + AVAILABLE_GATEWAY.keySet(), Application.SERVER_NAME_ID);

//                        LOGGER.info("{}重申对当前持有通道的占有权", Application.SERVER_NAME_ID);
                        claimDbHoldingPrivilege();//重申通道占用权
//                        LOGGER.info("{}释放长时间未进行发送的通道", Application.SERVER_NAME_ID);
                        releaseGatewayIfNecessary();//释放过期通道

//                        LOGGER.info("{}刷新当前存在通道的配置", Application.SERVER_NAME_ID);
                        syncGatewayConfigFromCacheDB();//从缓存同步通道配置
//                        LOGGER.info("{}增加通道的工作线程数", Application.SERVER_NAME_ID);
                        increaseWorkingThreadIfNeed();//如果仍有许多任务,那么增加工作线程

                        LOGGER.info("{}定时维护应用通道信息结束. " + AVAILABLE_GATEWAY.entrySet().stream().map(e -> "[" + e.getKey() + ":" + e.getValue().getMaxConcurrentThreadNum() + ":" + e.getValue().getCurrentRunningTreadNum() + "]").collect(Collectors.toList()), Application.SERVER_NAME_ID);
                    } catch (Exception e) {
                        LOGGER.error("{}维护应用通道信息异常", Application.SERVER_NAME_ID, e);
                    }
                    try {
                        TimeUnit.SECONDS.sleep(20);
                    } catch (InterruptedException e) {
                        LOGGER.error("{}维护应用通道信息被唤醒", Application.SERVER_NAME_ID, e);
                    }
                }
            }).start();
        }

        /**
         * 维持通道占有, 通过刷新占有时间 和 数据库时间
         */
        private static void claimDbHoldingPrivilege() {
            iteratorGatewayCats((gid, unit) -> {
//                LOGGER.info("{}开始更新数据库通道[{}]的占用时间", Application.SERVER_NAME_ID, unit.getGatewayId());
                gatewayOfficer.channelService.updateGatewayHolding(new HashMap<String, Object>() {{
                    put("gatewayId", unit.getGatewayId());
                    put("serId", unit.getPrivilegeSerId());
                    put("holdingTime", new Timestamp(System.currentTimeMillis()));
                }});
                return unit;
            });
        }

        /**
         * 通道空闲过久,释放通道占有, 当且仅当线程是0 且最后更新时间距离当前已经超过一定时间
         */
        private static void releaseGatewayIfNecessary() {
            iteratorGatewayCats((gid, unit) -> {
//                LOGGER.info("{}开始检查并尝试释放通道[{}]的占有权", Application.SERVER_NAME_ID, unit.getGatewayId());
                if (unit.getCurrentRunningTreadNum() == 0 && ((System.currentTimeMillis() - unit.getLastUpdateTime()) > RELEASE_GATEWAY_EXPIRE_MINUTE * 60 * 1000)) {
                    gatewayOfficer.channelService.releaseGatewayHolding(new HashMap<String, Object>() {{
                        put("gatewayId", unit.getGatewayId());
                        put("serId", unit.getPrivilegeSerId());
                    }});
                    LOGGER.info("释放通道[{}]的占有权完成", unit.getGatewayId());
                    return null;
                }
                return unit;
            });
        }

        /**
         * 如果线程过少,增加工作线程
         */
        private static void increaseWorkingThreadIfNeed() {
            iteratorGatewayCats((gid, unit) -> {
//                LOGGER.info("{}开始检查是否增加通道线程,[{}]当前工作线程数[{}],最大并行线程数[{}]", Application.SERVER_NAME_ID, unit.getGatewayId(), unit.getCurrentRunningTreadNum(), unit.getMaxConcurrentThreadNum());
                if (unit.getCurrentRunningTreadNum() > 0) {
                    syncSubmitWorkThreadIfPermitted(unit);
                }
                return unit;
            });
        }

        /**
         * 同步通道的配置
         * <p>
         * 1. 通道是否被时间限制发送
         * 2. 通道是否被有效性限制发送
         * 3. 通道最大线程数更新
         */
        private static void syncGatewayConfigFromCacheDB() {
            iteratorGatewayCats((gid, unit) -> {
//                LOGGER.info("{}开始从缓存数据库刷新通道[{}]的配置信息,包括:通道有效性,通道时效性,最大并行数,最大任务获取数和发送配置", Application.SERVER_NAME_ID, gid);
                Map<String, Object> dbGatewayConfig = gatewayOfficer.cacheService.getSendConfigCache(String.valueOf(gid));
                if (dbGatewayConfig != null) {//不存在说明 通道是无效的 sql里查询时已经带 isValid=0 条件
                    unit.setSendHandleAllowed(true);
                    unit.setSendTimeAllowed(sendTimeAllowed(dbGatewayConfig));
                    unit.setMaxConcurrentThreadNum((Integer) dbGatewayConfig.getOrDefault("threadnum", 1));
                    unit.setMaxTaskFetchSize((Integer) dbGatewayConfig.getOrDefault("limitnum", 20));
                    unit.setSendConfig((String) dbGatewayConfig.get("sendConfig"));
                    LOGGER.info("从数据库刷新发送配置:{}", unit);
                } else {
                    unit.setSendHandleAllowed(false);
                }
                return unit;
            });
        }

        private static void iteratorGatewayCats(BiFunction<Integer, GatewayOperator, GatewayOperator> gatewayOperation) {
            AVAILABLE_GATEWAY.forEachKey(PARALLELISM_NUM, (gid) -> AVAILABLE_GATEWAY.computeIfPresent(gid, gatewayOperation));
        }

    }

    /**
     * 带有锁机制的多线程发送
     * <p>
     * 1. 通道第一次进入的时候 新增线程
     * 2. 一直在发送 控制器 新增线程
     * 3. 在发送完成后将要退出时 如果有新任务进入, 再提交一次
     */
    private static void syncSubmitWorkThreadIfPermitted(GatewayOperator operator) {
        operator.setLastUpdateTime();
        synchronized (operator.syncBeforeSubmitThread()) {
            long threadId = Thread.currentThread().getId();
            LOGGER.info("线程{}[尝试]新增{}通道的发送线程", threadId, operator.getGatewayId());
            if (operator.getCurrentRunningTreadNum() < operator.getMaxConcurrentThreadNum()) {
                LOGGER.info("线程{}[确认]新增{}通道的发送线程", threadId, operator.getGatewayId());
                GLOBAL_GATEWAY_TASK_EXECUTOR.submit(new GatewayOperator.GatewayWorkerRunner(operator));
                operator.increaseCurrentRunningTreadNum();
            }
            LOGGER.info("线程{}[结束]新增{}通道的发送线程,当前线程数[{}],允许最大线程数[{}]", threadId, operator.getGatewayId(), operator.getCurrentRunningTreadNum(), operator.getMaxConcurrentThreadNum());
        }
    }

    private static boolean sendTimeAllowed(Map<String, Object> gatewayCacheMap) {
        boolean sendAllowed = true;
        Time stopBeginTime = gatewayCacheMap.get("stopBeginTime") != null ? (Time) gatewayCacheMap.get("stopBeginTime") : null;
        Time stopEndTime = gatewayCacheMap.get("stopEndTime") != null ? (Time) gatewayCacheMap.get("stopEndTime") : null;

        if (stopBeginTime != null && stopEndTime != null) {
            Date stopBeginDateTime = DateUtil.StrToDate(DateUtil.getTime("yyyy-MM-dd") + " " + DateUtil.getTime(stopBeginTime, "HH:mm:ss"), "yyyy-MM-dd HH:mm:ss");
            Date stopEndDateTime = DateUtil.StrToDate(DateUtil.getTime("yyyy-MM-dd") + " " + DateUtil.getTime(stopEndTime, "HH:mm:ss"), "yyyy-MM-dd HH:mm:ss");
            try {
                Date nowTime = new Date();
                if (stopBeginDateTime.before(stopEndDateTime)) {
                    if (nowTime.after(stopBeginDateTime) && nowTime.before(stopEndDateTime)) {
                        sendAllowed = false;
                    }
                } else {
                    if (nowTime.after(stopBeginDateTime) || nowTime.before(stopEndDateTime)) {
                        sendAllowed = false;
                    }
                }
            } catch (Exception e) {
                LOGGER.error("系统异常", e);
            }
        }
        return sendAllowed;
    }

}

